package com.hisense.ovcloud.mqtt.business.service;

import cn.hutool.json.JSONUtil;
import com.hisense.ovcloud.commons.dto.common.mqtt.MqttCtlMsg;
import com.hisense.ovcloud.mqtt.business.config.Constants;
import com.hisense.ovcloud.mqtt.business.config.MqttClientConfig;
import io.micrometer.common.util.StringUtils;
import io.vertx.core.buffer.Buffer;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.annotation.DependsOn;
import org.springframework.core.annotation.Order;
import org.springframework.kafka.core.reactive.ReactiveKafkaConsumerTemplate;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.nio.charset.StandardCharsets;

@Service
@Slf4j
@Order
@DependsOn("mqttCustomClient")
public class ReactiveConsumerService implements CommandLineRunner, InitializingBean {

    private final MqttClientConfig mqttClientConfig;

    private final AuthService authService;

    private static Scheduler ctlWorkerPool;

    private final MQTTClientConnector mqttClientConnector;

    final
    ReactiveKafkaConsumerTemplate<String, String> kafkaConsumerTemplate;

    public ReactiveConsumerService(ReactiveKafkaConsumerTemplate<String, String> kafkaConsumerTemplate, MqttClientConfig mqttClientConfig, AuthService authService, MQTTClientConnector mqttClientConnector) {
        this.kafkaConsumerTemplate = kafkaConsumerTemplate;
        this.mqttClientConfig = mqttClientConfig;
        this.authService = authService;
        this.mqttClientConnector = mqttClientConnector;
    }

    private Flux<Mono<Boolean>> msgConsummer() {
        return kafkaConsumerTemplate
                .receiveAutoAck()
                .map(this::handler);
    }


    protected Mono<Boolean> handler(ConsumerRecord<String, String> consumerRecord) {
        try {
            if (StringUtils.isNotEmpty(consumerRecord.value())) {
                String receivedTxt = consumerRecord.value().strip().replaceAll("\n", "");
                if (consumerRecord.topic().equals(mqttClientConfig.getCtlKafkaTopic())) {
                    MqttCtlMsg ctlMsg = JSONUtil.toBean(receivedTxt, MqttCtlMsg.class);
                    authService.getWifiOnlineInRedisDto(ctlMsg.getWifiId())
                            .zipWhen(wifiOnlineInRedisDto -> {
                                String topic = String.format("%s/%s/%s/queueMode/set/",
                                        StringUtils.isNotEmpty(ctlMsg.getType()) && ctlMsg.getType().equals(Constants.MQTT.MsgTransTypes.TT) ? ctlMsg.getMajorDomainId() : wifiOnlineInRedisDto.getDomain(),
                                        StringUtils.isNotEmpty(ctlMsg.getType()) && ctlMsg.getType().equals(Constants.MQTT.MsgTransTypes.TT) ? ctlMsg.getSubDomainId() : wifiOnlineInRedisDto.getSubDomain(),
                                        ctlMsg.getWifiId());
                                return Mono.just(mqttClientConnector.send(topic,
                                        Buffer.buffer(java.util.Base64.getDecoder().decode(ctlMsg.getPayload().getBytes(StandardCharsets.UTF_8)))));
                            }, (x, y) -> y)
                            .subscribeOn(ctlWorkerPool)
                            .subscribe();
                }
                return Mono.just(true);
            }
        } catch (Exception e) {
            log.error("consuming encountered error: {}", e.getMessage());
        }
        return Mono.just(true);
    }

    @Override
    public void run(String... args) {
        msgConsummer().subscribe(Mono::subscribe);
    }

    @Override
    public void afterPropertiesSet() {
        ctlWorkerPool = Schedulers.newBoundedElastic(20, 5000, "ctlWorkerPool");
    }

    public static void main(String[] args) {
        System.out.println(java.util.Base64.getEncoder().encode("A9kPQJZmAAAAAAMAJAAAyrc=".getBytes(StandardCharsets.UTF_8)));
    }
}

